package cn.gyouzhe.component.disruptor;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

/**
 * desc
 *
 * @author wangchuan
 * @createTime 2018/2/8 20:02
 */
@Slf4j
public class DisruptorComponent {
    private Disruptor<Event> disruptor;
    private RingBuffer<Event> ringBuffer;

    DisruptorComponent(Builder builder) {
        disruptor = new Disruptor<>(Event::new, builder.ringBufferSize, DaemonThreadFactory.INSTANCE, ProducerType.MULTI, new SleepingWaitStrategy());
        disruptor.setDefaultExceptionHandler(new DisruptorDefaultException());
        EventHandlerGroup<Event> eventHandlerGroup = disruptor.handleEventsWith(builder.eventHandlers);
        List<EventHandler<Event>[]> thenHandlerList = builder.thenHandlerList;
        if(null != thenHandlerList) {
            for(EventHandler<Event>[] eventHandlers : thenHandlerList) {
                eventHandlerGroup = eventHandlerGroup.then(eventHandlers);
            }
        }
        if(null != builder.workHandlers) {
            disruptor.handleEventsWithWorkerPool(builder.workHandlers);
        }
        ringBuffer = disruptor.start();
    }

    /**
     * 发布事件
     * @param data
     */
    public void publish(Object data) {
        long seq = ringBuffer.next();
        try {
            ringBuffer.get(seq).setData(data);
        } finally {
            ringBuffer.publish(seq);
        }

    }

    /**
     * 关闭分发器
     */
    public void shutdown() {
        disruptor.shutdown();
    }

    /**
     * 创建builder
     * @return
     */
    public final static Builder create() {
        return new Builder();
    }

    public interface Wraper {
        /**
         * 任意一个处理器执行即可
         * @param workHandlers
         * @return
         */
        Wraper workHandlers(WorkHandler<Event>... workHandlers);

        /**
         * handler有序执行
         * @param eventEventHandlers
         * @return
         */
        Wraper then(EventHandler<Event>... eventEventHandlers);

        /**
         * 构建组件
         * @return
         */
        DisruptorComponent build();


    }

    public static class Builder implements Wraper {
        private int ringBufferSize = 2048;
        private EventHandler<Event>[] eventHandlers;
        private WorkHandler<Event>[] workHandlers;
        private List<EventHandler<Event>[]> thenHandlerList;

        private Builder() {
        }

        /**
         * 获取往后的处理器
         * @return
         */
        private List<EventHandler<Event>[]> getThenHandlerList() {
            if(null == thenHandlerList) {
                thenHandlerList = new ArrayList<>();
            }
            return thenHandlerList;
        }

        /**
         * 设置队列大小
         * @param size
         * @return
         */
        public final Builder ringBufferSize(int size) {
            if(Integer.bitCount(size) != 1) {
                throw new RuntimeException("buffer大小必须为2的幂指数");
            }
            this.ringBufferSize = size;
            return this;
        }

        /**
         * 并行
         * @param eventHandlers
         */
        public Wraper handlers(EventHandler<Event>... eventHandlers) {
            this.eventHandlers = eventHandlers;
            return this;
        }

        /**
         * 任意选择处理器执行
         * @param workHandlers
         * @return
         */
        @Override
        public Wraper workHandlers(WorkHandler<Event>... workHandlers) {
            this.workHandlers = workHandlers;
            return this;
        }

        /**
         * 后续执行器
         * @param eventEventHandlers
         * @return
         */
        @Override
        public Wraper then(EventHandler<Event>... eventEventHandlers) {
            getThenHandlerList().add(eventEventHandlers);
            return this;
        }

        public DisruptorComponent build() {
            if(null == this.eventHandlers) {
                throw new RuntimeException("事件执行器不能为空");
            }
            return new DisruptorComponent(this);
        }

    }
}
